/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */

package cgl.narada.wsinfra.wsr.impl;

import java.util.Calendar;
import java.util.Hashtable;
import java.util.Vector;

import javax.xml.namespace.QName;

import cgl.narada.wsinfra.WsMessageFlow;
import cgl.narada.wsinfra.WsProcessor;
import cgl.narada.wsinfra.exception.DeploymentException;
import cgl.narada.wsinfra.exception.IncorrectExchangeException;
import cgl.narada.wsinfra.exception.MessageFlowException;
import cgl.narada.wsinfra.exception.ProcessingException;
import cgl.narada.wsinfra.exception.UnknownExchangeException;
import cgl.narada.wsinfra.exception.WsFaultException;
import cgl.narada.wsinfra.schemas.addressing.EndpointReferenceType;
import cgl.narada.wsinfra.schemas.addressing.ToDocument;
import cgl.narada.wsinfra.schemas.soap.EnvelopeDocument;
import cgl.narada.wsinfra.schemas.wsr.RequestDocument;
import cgl.narada.wsinfra.schemas.wsr.ResponseDocument;
import cgl.narada.wsinfra.util.SoapMessageAlteration;
import cgl.narada.wsinfra.wsa.AddressingHeaders;
import cgl.narada.wsinfra.wsa.ParseWsaHeaders;
import cgl.narada.wsinfra.wsa.WsaEprRules;
import cgl.narada.wsinfra.wsa.WsaProcessingFactory;
import cgl.narada.wsinfra.wsr.WsrAckOperations;
import cgl.narada.wsinfra.wsr.WsrElementAddition;
import cgl.narada.wsinfra.wsr.WsrElementCreation;
import cgl.narada.wsinfra.wsr.WsrExchangeInfo;
import cgl.narada.wsinfra.wsr.WsrExchangeInfoCreator;
import cgl.narada.wsinfra.wsr.WsrNodeUtils;
import cgl.narada.wsinfra.wsr.WsrProcessingFactory;
import cgl.narada.wsinfra.wsr.WsrQNames;
import cgl.narada.wsinfra.wsr.WsrRequestCreator;
import cgl.narada.wsinfra.wsr.WsrSource;
import cgl.narada.wsinfra.wsr.storage.WsrGroupInfoStorageOperations;
import cgl.narada.wsinfra.wsr.storage.WsrProtocolStorageOperations;
import cgl.narada.wsinfra.wsr.storage.WsrStorageException;
import cgl.narada.wsinfra.wsr.storage.WsrStorageService;
import cgl.narada.wsinfra.wsr.storage.impl.WsrGroupInfoImpl;
import cgl.narada.wsinfra.wsr.storage.impl.WsrGroupTerminationInfoImpl;
import cgl.narada.wsinfra.wsr.storage.impl.WsrStorageWidgetImpl;


/**This class implements the functionality of a WSR Source.

   @author Sima Patel
   $Date: 2005/05/04 18:05:34 $
   $Revision: 1.12 $*/

public class WsrSourceNode extends WsProcessor implements WsrSource {

    private EndpointReferenceType sourceEpr;
    private ParseWsaHeaders parseWsaHeaders;
    private WsaEprRules wsaEprRules;
    private WsMessageFlow wsMessageFlow;
    private WsrSourceNodeHelper sourceNodeHelper;
    private WsrExchangeInfoCreator wsrExchangeInfoCreator;
    private WsrElementCreation wsrElementCreation;
    private WsrElementAddition wsrElementAddition;
    private WsrNodeUtils wsrNodeUtils;
    private WsrQNames wsrQNames;
    private WsrAckOperations wsrAckOperations;
    private WsrUtil wsrUtil;
    private WsrFaultsImpl wsrFaultsImpl;
    private WsrRequestCreator wsrRequestCreator;
    private WsrSourceGroupTerminationImpl wsrSourceGroupTerminationImpl;
    private WsrGroupInfoStorageOperations wsrGroupInfoStorageOperations;
    private WsrProtocolStorageOperations wsrProtocolStorageOperations;
    private WsrStorageService wsrStorageServiceSource;
    private WsrGroupInfoImpl wsrGroupInfoImpl;
    private WsrGroupTerminationInfoImpl wsrGroupTerminationInfoImpl;
    private WsrStorageWidgetImpl wsrStorageWidgetImpl;
    private SoapMessageAlteration soapMessageAlteration;
    public static int counter =0;


    private boolean isNewGroup = false;
    private String moduleName = "WsrSourceNode: ";

    public WsrSourceNode(String configInfo) throws DeploymentException{
    	try {
          wsrStorageServiceSource = WsrStorageService.getSourceStorageServiceInstance(configInfo);
          wsrProtocolStorageOperations =
                           wsrStorageServiceSource.getWsrProtocolStorageOperations();
          wsrGroupInfoStorageOperations =
                           wsrStorageServiceSource.getWsrGroupInfoStorageOperations();
        } catch (WsrStorageException wsrStorageException) {
          String reason = moduleName + "Problems initializing storage services"
            + ", please check the location of the config file [" + configInfo +
            "] that was provided.\n" + wsrStorageException.toString();
          throw new DeploymentException(reason);
        }

        parseWsaHeaders = WsaProcessingFactory.getParseWsaHeaders();
        wsaEprRules = WsaProcessingFactory.getWsaEprRules();
        soapMessageAlteration = SoapMessageAlteration.getInstance();
        wsrFaultsImpl = WsrFaultsImpl.getInstance();
        wsrQNames = WsrQNames.getInstance();
        wsrElementCreation = WsrProcessingFactory.getWsrElementCreation();
        wsrElementAddition = WsrProcessingFactory.getWsrElementAddition();
        wsrExchangeInfoCreator = WsrProcessingFactory.getWsrExchangeInfoCreator();
        wsrAckOperations = WsrProcessingFactory.getWsrAckOperations();
        wsrRequestCreator = WsrProcessingFactory.getWsrRequestCreator();
        wsrNodeUtils = WsrProcessingFactory.getWsrNodeUtils();
        wsrUtil = WsrUtil.getInstance();
        sourceNodeHelper = WsrSourceNodeHelper.getInstance();
        sourceNodeHelper.initialize(wsrStorageServiceSource);
        wsrSourceGroupTerminationImpl = WsrSourceGroupTerminationImpl.getInstance();
        wsrSourceGroupTerminationImpl.initialize(wsrStorageServiceSource);
	
    }

    /** Sets the message flow which the processor should use */
    public void setMessageFlow(WsMessageFlow wsMessageFlow)
                throws DeploymentException {
        this.wsMessageFlow = wsMessageFlow;
        sourceNodeHelper.setMessageFlow(wsMessageFlow);
    }

    /** Gets the message flow which the processor should use. */
    public WsMessageFlow getMessageFlow() {
        return wsMessageFlow;
    }

   /** Retrieves the endpoint reference associated with this node */
    public EndpointReferenceType getEndpointReference() {
        return sourceEpr;
    }

    /** Set the endpoint reference associated with this node */
    public void setEndpointReference(EndpointReferenceType endpointReference) {
        sourceEpr = endpointReference;
        sourceNodeHelper.setEndpointReference(sourceEpr);
    }

    /** Process the exchange. The arguments also indicate the direction in
    which the exchange has actually traversed and the destination address */
    public boolean processExchange(EnvelopeDocument  envelopeDocument,
			                    int direction)
                throws UnknownExchangeException, IncorrectExchangeException,
                ProcessingException, MessageFlowException {


        WsrExchangeInfo wsrExchangeInfo =
            wsrExchangeInfoCreator.createWsrExchangeInfo(envelopeDocument);
        AddressingHeaders addressingHeaders =
            wsrExchangeInfo.getAddressingHeaders();
        boolean isFault = isFault(addressingHeaders);
        if (isFault) {
          return true;
        }

        sourceNodeHelper.checkExchangeType(wsrExchangeInfo, direction);


        try{

          /** Only valid exchanges to be processed now remain */
          if (direction == WsMessageFlow.FROM_APPLICATION) {

              if(wsrExchangeInfo.isProcessRequest()){

                processMessageFromApplication(envelopeDocument,addressingHeaders);
              }

          return true;
          }
          /** From here on every thing else has been received over the network */
          if(!wsrExchangeInfo.hasResponse()){
              enrouteToApplication(envelopeDocument);
              return false;
          }
          processResponse(envelopeDocument, addressingHeaders);
          return false;
          }catch (WsFaultException wsFaultException) {
          manageFaultException(wsFaultException, direction);
          }catch (WsrStorageException wsrStorageException) {
          /** Need to manage storage exception here */
          System.out.println(moduleName + wsrStorageException.toString() );
          }
          return true;
    }//end of processexchange method


    /** Process exchange received from application, and send it across
    reliability */
    public void processMessageFromApplication(EnvelopeDocument envelopeDocument,
                                            AddressingHeaders addressingHeaders)
                throws WsFaultException,  MessageFlowException,
                       WsrStorageException{

        String groupId = null;
        sourceNodeHelper.checkForProblemsInToDocument(addressingHeaders);
        ToDocument toDocument = addressingHeaders.getTo();
        String destination = toDocument.getTo().getStringValue();

        groupId = wsrNodeUtils.getGroupId(envelopeDocument);
        if(groupId == null){
        	groupId = retrieveGroupId(destination);
        }else{

            if(!wsrGroupInfoStorageOperations.isKnownGroup(groupId)){
                isNewGroup = true;
            }else{

                String destinationaddress
                       = wsrGroupInfoStorageOperations.getDestinationForGroupId
                                                                      (groupId);

                isNewGroup=false;
                if(!destinationaddress.equalsIgnoreCase(destination)){
                  String error = "You can not specify same groupId " +
                          "for multiple destinations";
                  WsFaultException wsFaultException =
                    new WsFaultException(wsrFaultsImpl.getSoapSenderQName(),
                                                                         error);
                  wsFaultException.setFaultTo(sourceEpr);
                  throw wsFaultException;
                }
            }
        }
        if(isNewGroup){

           if(!processNewGroup(groupId, destination, envelopeDocument)){
                String errorReport = "UnSuccessFul processing of new group, " +
                        "because either expiry time passed or group is closed ";
                  WsFaultException wsFaultException =
                        new WsFaultException(wsrFaultsImpl.getSoapSenderQName(),
                                                                  errorReport);
                  wsFaultException.setFaultTo(sourceEpr);
                  throw wsFaultException;
           }
        }
        else{

           if(!processExistingGroup(groupId,envelopeDocument)){
               String errorReport = "UnSuccessFul processing of " +
                       "Existing group, because group is closed or " +
                       "expirytime passed";
                  WsFaultException wsFaultException =
                        new WsFaultException(wsrFaultsImpl.getSoapSenderQName(),
                                                                  errorReport);
                  wsFaultException.setFaultTo(sourceEpr);
                  throw wsFaultException;
           }
        }
        removeHeadersFromEnvelopeDocument(envelopeDocument);
        enrouteToNetwork(envelopeDocument);

        RequestDocument requestDocument = wsrNodeUtils.getRequestDocument
                                                            (envelopeDocument);
        if(requestDocument.getRequest().getReplyPattern().getValue().toString().
                                               trim().equalsIgnoreCase("POLL")){

            createAndEnroutePollRequest(toDocument);

        }

    }
    /** Remove extra headers from EnvelopeDocument*/
    public void removeHeadersFromEnvelopeDocument(EnvelopeDocument
                                                              envelopeDocument){
    	QName qNameGroupId = wsrQNames.getGroupId();
    	boolean removedGroupId=
          soapMessageAlteration.removeFromSoapHeader(envelopeDocument,
                  qNameGroupId);


        QName qNameReplyPattern = wsrQNames.getRPattern();
    	boolean removedReplyPattern =
          soapMessageAlteration.removeFromSoapHeader(envelopeDocument,
                  qNameReplyPattern);


        QName qNameReplyTo = wsrQNames.getReplyTo();
    	boolean removedReplyTo =
          soapMessageAlteration.removeFromSoapHeader(envelopeDocument,
                  qNameReplyTo);


        QName qNameLastMessage = wsrQNames.getLastMessage();
    	boolean removedLastMessage =
          soapMessageAlteration.removeFromSoapHeader(envelopeDocument,
                  qNameLastMessage);




    }
   /** Creates PollRequestDocument from specified parameters and
    enroute to the network */
    public void createAndEnroutePollRequest(ToDocument toDocument)
                throws WsrStorageException, MessageFlowException{

        Hashtable groupIdAndRangesTable = null;
        String destination = toDocument.getTo().getStringValue();
        String[] groupIds =
           wsrGroupInfoStorageOperations.getGroupIdsForDestination(destination);
        if(groupIds == null || groupIds.length == 0){
            return;
        }
        groupIdAndRangesTable = new Hashtable();
        for(int index = 0; index < groupIds.length; index++){
            String groupId = groupIds[index];
            long[] unAckSequenceNumbers =
               wsrProtocolStorageOperations.getUnacknowledgedSequenceNumbers
                                                              (groupIds[index]);
            if(unAckSequenceNumbers == null || unAckSequenceNumbers.length == 0){
                continue;
            }
            Hashtable ranges = wsrAckOperations.createRanges(unAckSequenceNumbers);
            groupIdAndRangesTable.put(groupId,ranges);
         }
        EnvelopeDocument envelopeDocument = wsrRequestCreator.getPollRequest
                                 (toDocument, sourceEpr,groupIdAndRangesTable);
        if(envelopeDocument != null){
          enrouteToNetwork(envelopeDocument);
        }
    }

    /** Process New Group Request */
    public boolean processNewGroup(String groupId, String destination,
                                   EnvelopeDocument envelopeDocument)
                   throws WsrStorageException,WsFaultException{

        EndpointReferenceType destinationAddress =
                                    EndpointReferenceType.Factory.newInstance();
        destinationAddress.addNewAddress().setStringValue(destination);
        //Creates GroupInfo Object
        wsrGroupInfoImpl = createGroupInfo(groupId, envelopeDocument);
        wsrGroupInfoImpl.setIsSource(true);
        wsrGroupInfoImpl.setDestination(destinationAddress);
        wsrGroupInfoImpl.setSource(sourceEpr);

        //Create StorageWidget Object
        wsrStorageWidgetImpl = createMessageInfo(groupId,0,envelopeDocument);

        //check if messageExpiryTime is < groupExpirytime or
        //groupMaxIdleDuration is if present
        if(wsrGroupInfoImpl.hasGroupExpiryTime()){
          if(wsrStorageWidgetImpl.getMessageExpiryTime().getTimeInMillis() >
             wsrGroupInfoImpl.getGroupExpiryTime().getTimeInMillis()){
         	return false;
         }

        }
        //Create GroupTerminationInfo Object
        wsrGroupTerminationInfoImpl = new WsrGroupTerminationInfoImpl(groupId);

        wsrGroupTerminationInfoImpl.setIsSource(true);
        if(wsrGroupInfoImpl.getGroupExpiryTime() != null){
        	if(wsrUtil.isExpiryTimePassed(wsrGroupInfoImpl.getGroupExpiryTime())){
        		return false;
        	}

        }
        if(wsrGroupInfoImpl.getGroupMaxIdleDuration() != null){
        	if(wsrUtil.isDurationPassed(wsrGroupInfoImpl.getGroupMaxIdleDuration())){
        		return false;
        	}

        }
        if(wsrUtil.isExpiryTimePassed(wsrStorageWidgetImpl.getMessageExpiryTime())){
        	return false;
        }
        if(wsrNodeUtils.getLastMessage(envelopeDocument) != null &&
        	wsrNodeUtils.getLastMessage(envelopeDocument).equalsIgnoreCase("TRUE")){

            wsrGroupInfoImpl.setLastSequenceNumber(0);
            wsrGroupTerminationInfoImpl.setLastMessageOfGroup(true);
        }
         /** create request from parameters */
        RequestDocument requestDocument = sourceNodeHelper.createRequest(
                                 groupId,wsrGroupInfoImpl.getGroupExpiryTime(),
                                 wsrGroupInfoImpl.getGroupMaxIdleDuration(), 0,
                                 wsrGroupInfoImpl.hasLastSequenceNumber() ,
                                 wsrStorageWidgetImpl.getMessageExpiryTime(),
								 wsrStorageWidgetImpl.getReplyPattern(),
                                 wsrStorageWidgetImpl.getReplyTo());
        wsrElementAddition.addRequest(envelopeDocument, requestDocument);
        wsrStorageWidgetImpl.setEnvelope(envelopeDocument);
        wsrGroupInfoStorageOperations.addDestinationGroupIdPair(
                                                          destination, groupId);
        wsrProtocolStorageOperations.store(wsrStorageWidgetImpl);
        wsrProtocolStorageOperations.storeGroupTerminationInfo
                                                  (wsrGroupTerminationInfoImpl);
        wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
        return true;

    }
    /** process Existing Group Request */
    public boolean processExistingGroup(String groupId,
                                        EnvelopeDocument envelopeDocument)
                   throws WsrStorageException, WsFaultException{

        long sequenceNumber = 0;
        wsrGroupTerminationInfoImpl =
                  wsrProtocolStorageOperations.getGroupTerminationInfo(groupId);
        if(wsrGroupTerminationInfoImpl.isGroupClosed()){

          return false;
        }

        wsrGroupInfoImpl =
            wsrGroupInfoStorageOperations.getGroupInfoUsingGroupId(groupId);
        long previousSequenceNumber =
            wsrProtocolStorageOperations.getSequenceNumberLastAssigned(groupId);
        sequenceNumber = previousSequenceNumber + 1;

        wsrStorageWidgetImpl = createMessageInfo(groupId, sequenceNumber,
        		                                 envelopeDocument);
        if(wsrGroupInfoImpl.getGroupExpiryTime() != null){
        	if(wsrUtil.isExpiryTimePassed(wsrGroupInfoImpl.getGroupExpiryTime())){
        		return false;
        	}

        }
        if(wsrGroupInfoImpl.getGroupMaxIdleDuration() != null){
        	if(wsrUtil.isDurationPassed(wsrGroupInfoImpl.getGroupMaxIdleDuration())){
        		return false;
        	}

        }
        if(wsrUtil.isExpiryTimePassed(wsrStorageWidgetImpl.getMessageExpiryTime())){
        	return false;
        }
        if(wsrNodeUtils.getLastMessage(envelopeDocument) != null &&
           wsrNodeUtils.getLastMessage(envelopeDocument).equalsIgnoreCase("TRUE")){
            wsrGroupInfoImpl.setLastSequenceNumber(sequenceNumber);
            wsrGroupTerminationInfoImpl.setLastMessageOfGroup(true);
        }
        if (sequenceNumber == Long.MAX_VALUE){
            wsrGroupTerminationInfoImpl.setMaxSequenceNumber(true);
        }
        RequestDocument requestDocument = sourceNodeHelper.createRequest(
                                 groupId,wsrGroupInfoImpl.getGroupExpiryTime(),
                                 wsrGroupInfoImpl.getGroupMaxIdleDuration(),
                                 sequenceNumber,
                                 wsrGroupInfoImpl.hasLastSequenceNumber(),
                                 wsrStorageWidgetImpl.getMessageExpiryTime(),
                                 wsrStorageWidgetImpl.getReplyPattern(),
                                 wsrStorageWidgetImpl.getReplyTo());

        wsrElementAddition.addRequest(envelopeDocument, requestDocument);
        wsrStorageWidgetImpl.setEnvelope(envelopeDocument);

        wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
        wsrProtocolStorageOperations.storeGroupTerminationInfo
        (wsrGroupTerminationInfoImpl);
        wsrProtocolStorageOperations.store(wsrStorageWidgetImpl);
        return true;

    }

    /** create GroupInfo from Property file*/
    public WsrGroupInfoImpl createGroupInfo(String groupId,
    		                                EnvelopeDocument envelopeDocument)
                            throws WsFaultException{
        String groupExpiryTime =
                              wsrNodeUtils.getGroupExpiryTime(envelopeDocument);
    	String groupMaxIdleDuration =
    	                   wsrNodeUtils.getGroupMaxIdleDuration(envelopeDocument);
        WsrGroupInfoImpl wsrGroupInfoImpl = new WsrGroupInfoImpl(groupId);
        if(groupExpiryTime != null && groupMaxIdleDuration != null){
            wsrGroupInfoImpl.setGroupExpiryTime
                                 (wsrUtil.getCalendarOfString(groupExpiryTime));
            return wsrGroupInfoImpl;

        }else if(groupExpiryTime != null){
            wsrGroupInfoImpl.setGroupExpiryTime(
            		wsrUtil.getCalendarOfString(groupExpiryTime));
            return wsrGroupInfoImpl;
        }else if(groupMaxIdleDuration != null){
            wsrGroupInfoImpl.setGroupMaxIdleDuration(
                               wsrUtil.getDurationOfString(groupMaxIdleDuration));
            return wsrGroupInfoImpl;
        }else{
        	Calendar current = Calendar.getInstance();
            current.add(Calendar.MONTH, 2);
            wsrGroupInfoImpl.setGroupExpiryTime(current);
            return wsrGroupInfoImpl;
        }

    }
    /** Create MessageInfo from Property file */
    public WsrStorageWidgetImpl createMessageInfo(String groupId,
                                                  long sequenceNumber,
											      EnvelopeDocument envelopeDocument)
                                throws WsFaultException{

        WsrStorageWidgetImpl wsrStorageWidgetImpl = new WsrStorageWidgetImpl(
                                                       groupId, sequenceNumber);
        String messageExpiryTime = wsrNodeUtils.getMExpiryTime(envelopeDocument);
        if(messageExpiryTime != null){
        	wsrStorageWidgetImpl.setMessageExpiryTime(
        			wsrUtil.getCalendarOfString(messageExpiryTime));
        }else{
        	Calendar current = Calendar.getInstance();
        	current.add(Calendar.MONTH, 1);
        	wsrStorageWidgetImpl.setMessageExpiryTime(current);
        }
        String replyPattern = wsrNodeUtils.getRPattern(envelopeDocument);
        if(replyPattern != null){
        	wsrStorageWidgetImpl.setReplyPattern(replyPattern);
        }else{
        	wsrStorageWidgetImpl.setReplyPattern("Response");
        }
        if(wsrStorageWidgetImpl.getReplyPattern().trim().equalsIgnoreCase
                                                                 ("Callback")){
        	String replyTo = wsrNodeUtils.getReplyTo(envelopeDocument);
        	if(replyTo != null){
        		wsrStorageWidgetImpl.setReplyTo(replyTo);
        	}else{
        		wsrStorageWidgetImpl.setReplyTo(
        		                           sourceEpr.getAddress().getStringValue());
        	}
        }
        wsrStorageWidgetImpl.setRetransmissionTime(
                                             System.currentTimeMillis()+300000);
        return wsrStorageWidgetImpl;


    }

    /** Process group acknowledgements and faults received over the wire
    from the sink */
    public void processResponse(EnvelopeDocument envelopeDocument,
                                AddressingHeaders addressingHeaders)
                throws WsFaultException, MessageFlowException,
                       WsrStorageException{

        ResponseDocument responseDocument = wsrNodeUtils.getResponseDocument(
                                                              envelopeDocument);
        boolean isFaultsInNonSequenceReply =
                                      processNonSequenceReply(responseDocument);
        boolean isFaultsInSequenceReply =
                                      processSequenceReply(responseDocument);
        processTermination(responseDocument);
        if(isFaultsInNonSequenceReply || isFaultsInSequenceReply){
            enrouteToApplication(envelopeDocument);
        }
    }
    /** process Non Sequence Reply */
    public boolean processNonSequenceReply(ResponseDocument responseDocument)
                   throws WsrStorageException, MessageFlowException{

        boolean acknowledgements = false;
        boolean retransmissions = false;
        boolean faults = false;

        Vector nonSequenceReplyAcknowledgementVector =
          wsrAckOperations.getNonSequenceReplyAcknowledgement(responseDocument);
        if(nonSequenceReplyAcknowledgementVector != null){
	    if(nonSequenceReplyAcknowledgementVector.isEmpty()==false ){
            acknowledgements = true;
            sourceNodeHelper.processNonSequenceReplyAcknowledgement(
                                         nonSequenceReplyAcknowledgementVector);
	    }
        }
        Vector retransmissionVector =
            wsrAckOperations.getNonSequenceReplyMessageProcessingFailureFault(
                                                              responseDocument);
        if(retransmissionVector != null){
	    if(retransmissionVector.isEmpty()==false){
            retransmissions = true;
            sourceNodeHelper.processNonSequenceReplyMessageProcessingFailureFault
                                                         (retransmissionVector);
	    }
        }
        Hashtable nonSequenceReplyFault =
                    wsrAckOperations.getNonSequenceReplyFault(responseDocument);
        if(nonSequenceReplyFault != null){
	    if(nonSequenceReplyFault.isEmpty() == false){
            faults = true;
            sourceNodeHelper.processNonSequenceReplyFault(nonSequenceReplyFault);
	    }
        }
        if(acknowledgements == false && retransmissions == false &&
                                                            faults == true){
            return true;
        }else {
            return false;
        }

    }
    /** process Sequence Reply */
    public boolean processSequenceReply(ResponseDocument responseDocument)
                   throws WsrStorageException, MessageFlowException{

        boolean acknowledgements = false;
        boolean retransmissions = false;
        boolean faults = false;

        Hashtable sequenceRepliesAcknowledgement =
           wsrAckOperations.getSequenceRepliesAcknowledgement(responseDocument);
        if(sequenceRepliesAcknowledgement != null){
	    if(sequenceRepliesAcknowledgement.isEmpty()== false){
            acknowledgements = true;
            sourceNodeHelper.processSequenceReplyAcknowledgement(
                                                sequenceRepliesAcknowledgement);
	    }
        }
        Hashtable retransmissionTable =
               wsrAckOperations.getSequenceReplyMessageProcessingFailureFault(
                                                              responseDocument);
        if(retransmissionTable != null){
	    if(retransmissionTable.isEmpty() != true){
            retransmissions = true;
            sourceNodeHelper.processSequenceReplyMessageProcessingFailureFault(
                                                          retransmissionTable);
	    }
        }
        Hashtable sequenceReplyFaultTable =
                wsrAckOperations.getSequenceReplyFault(responseDocument);
        if(sequenceReplyFaultTable  != null){
	    if(sequenceReplyFaultTable.isEmpty()!= true){
            faults = true;
            sourceNodeHelper.processSequenceReplyFault(
                                                       sequenceReplyFaultTable);
	    }
        }
        if(acknowledgements == false && retransmissions == false &&
                                                           faults == true){
            return true;
        }else {
            return false;
        }

    }
    /** process Sequence Termination */
    public void processTermination(ResponseDocument responseDocument)
                throws WsrStorageException{
        String[] groupIds = wsrAckOperations.getGroupIds(responseDocument);
        for(int index =0; index<groupIds.length;index++){
            wsrSourceGroupTerminationImpl.processGroupTermination(groupIds[index]);
        }
    }

    /** Retrieves the GroupId associated with destinationAddress */
    private String retrieveGroupId(String destinationAddress)
                   throws WsrStorageException, WsFaultException{

        if(!wsrGroupInfoStorageOperations.hasGroupIdForDestination(
                                                           destinationAddress)){

            isNewGroup = true;
            return sourceNodeHelper.createGroupId();
        }else{
            String[] groupId =
                   wsrGroupInfoStorageOperations.getGroupIdsForDestination(
                                                            destinationAddress);
            if(groupId.length == 1){
            	WsrGroupTerminationInfoImpl wsrGroupTerminationInfoImpl1 =
            		wsrProtocolStorageOperations.getGroupTerminationInfo(groupId[0]);
            	if(wsrGroupTerminationInfoImpl1.isGroupClosed()){
            		isNewGroup = true;

            		return sourceNodeHelper.createGroupId();
            	}else{
	            	isNewGroup = false;
	            	return groupId[0];
            	}

            }else{
                String faultError = "More than one groupIds are " +
                        "exist for single Destination in database";
                WsFaultException wsFaultException = new WsFaultException(
                                 wsrFaultsImpl.getSoapSenderQName(),faultError);
                wsFaultException.setFaultTo(sourceEpr);
                throw wsFaultException;

            }
        }
    }

    private void reportError(String error) {
        System.out.println(moduleName + error);
    }
}
